/**
 * Software License, Version 1.0 Copyright 2003 The Trustees of Indiana
 * University. All rights reserved. Redistribution and use in source and binary
 * forms, with or without modification, are permitted provided that the
 * following conditions are met: 1) All redistributions of source code must
 * retain the above copyright notice, the list of authors in the original source
 * code, this list of conditions and the disclaimer listed in this license; 2)
 * All redistributions in binary form must reproduce the above copyright notice,
 * this list of conditions and the disclaimer listed in this license in the
 * documentation and/or other materials provided with the distribution; 3) Any
 * documentation included with all redistributions must include the following
 * acknowledgement: "This product includes software developed by the Community
 * Grids Lab. For further information contact the Community Grids Lab at
 * http://communitygrids.iu.edu/." Alternatively, this acknowledgement may
 * appear in the software itself, and wherever such third-party acknowledgments
 * normally appear. 4) The name Indiana University or Community Grids Lab or
 * NaradaBrokering, shall not be used to endorse or promote products derived
 * from this software without prior written permission from Indiana University.
 * For written permission, please contact the Advanced Research and Technology
 * Institute ("ARTI") at 351 West 10th Street, Indianapolis, Indiana 46202. 5)
 * Products derived from this software may not be called NaradaBrokering, nor
 * may Indiana University or Community Grids Lab or NaradaBrokering appear in
 * their name, without prior written permission of ARTI. Indiana University
 * provides no reassurances that the source code provided does not infringe the
 * patent or any other intellectual property rights of any other entity. Indiana
 * University disclaims any liability to any recipient for claims brought by any
 * other entity based on infringement of intellectual property rights or
 * otherwise. LICENSEE UNDERSTANDS THAT SOFTWARE IS PROVIDED "AS IS" FOR WHICH
 * NO WARRANTIES AS TO CAPABILITIES OR ACCURACY ARE MADE. INDIANA UNIVERSITY
 * GIVES NO WARRANTIES AND MAKES NO REPRESENTATION THAT SOFTWARE IS FREE OF
 * INFRINGEMENT OF THIRD PARTY PATENT, COPYRIGHT, OR OTHER PROPRIETARY RIGHTS.
 * INDIANA UNIVERSITY MAKES NO WARRANTIES THAT SOFTWARE IS FREE FROM "BUGS",
 * "VIRUSES", "TROJAN HORSES", "TRAP DOORS", "WORMS", OR OTHER HARMFUL CODE.
 * LICENSEE ASSUMES THE ENTIRE RISK AS TO THE PERFORMANCE OF SOFTWARE AND/OR
 * ASSOCIATED MATERIALS, AND TO THE PERFORMANCE AND VALIDITY OF INFORMATION
 * GENERATED USING SOFTWARE.
 */
package cgl.narada.discovery.topics;

import java.io.FileNotFoundException;
import java.io.FileOutputStream;
import java.io.IOException;
import java.security.InvalidKeyException;
import java.security.NoSuchAlgorithmException;
import java.security.NoSuchProviderException;
import java.security.PrivateKey;
import java.security.PublicKey;
import java.security.SignatureException;
import java.security.cert.Certificate;
import java.security.cert.CertificateException;
import java.security.cert.X509Certificate;
import java.util.Calendar;
import java.util.Enumeration;
import java.util.Hashtable;
import java.util.Properties;
import java.util.Random;
import java.util.regex.Matcher;
import java.util.regex.Pattern;

import org.apache.log4j.Logger;

import cgl.narada.discovery.broker.BrokerDiscoveryHelper;
import cgl.narada.discovery.broker.messages.BrokerDiscoveryResponse;
import cgl.narada.discovery.topics.messages.EncryptedMessagePayload;
import cgl.narada.discovery.topics.messages.SignedTopicAdvertisement;
import cgl.narada.discovery.topics.messages.TDNDiscoveryRequest;
import cgl.narada.discovery.topics.messages.TDNDiscoveryResponse;
import cgl.narada.discovery.topics.messages.TopicAdvertisement;
import cgl.narada.discovery.topics.messages.TopicCreationRequest;
import cgl.narada.discovery.topics.messages.TopicDiscoveryRequest;
import cgl.narada.discovery.topics.messages.TopicDiscoveryResponse;
import cgl.narada.event.NBEvent;
import cgl.narada.event.TemplateProfileAndSynopsisTypes;
import cgl.narada.matching.Profile;
import cgl.narada.service.ServiceException;
import cgl.narada.service.ServiceProperties;
import cgl.narada.service.ServicePropertiesFactory;
import cgl.narada.service.client.ClientService;
import cgl.narada.service.client.EventConsumer;
import cgl.narada.service.client.EventProducer;
import cgl.narada.service.client.NBEventListener;
import cgl.narada.service.client.SessionService;
import cgl.narada.service.security.securityprovider.CertificateManager;
import cgl.narada.service.security.securityprovider.CertificateUtil;
import cgl.narada.service.timer.ClockI;
import cgl.narada.service.timer.HRClock;
import cgl.narada.util.SystemInit;
import cgl.narada.util.SystemNotInitializedException;
import cgl.narada.util.UUIDRetriever;

/**
 * Implementation of a Topic Discovery Node Created on Apr 7, 2005
 * 
 * @author Harshawardhan Gadgil (hgadgil@grids.ucs.indiana.edu)
 */

public class TopicDiscoveryNode implements NBEventListener {

   static Logger log = Logger.getLogger("TopicDiscoveryNode");

   private String TDN_TOPIC;

   /** The ID of this TopicDiscoveryNode */
   private String TDN_ID;

   /** My Credentials identifying me */
   private Certificate TDN_Credentials;

   private PrivateKey privateKey;

   private String brokerHost;
   private String brokerPort;
   private String brokerConnProtocol;
   private String configFilePath;

   private String TDN_ID_TOPIC;

   private Hashtable storedTopicAds;

   private EventConsumer consumer;
   private EventProducer producer;
   private PublicKey rootPubKey;

   private ServiceProperties serviceProperties;

   // BECNHMARK
   private boolean benchmark = false;
   private ClockI hrClock = null;
   private FileOutputStream fout = null;

   public TopicDiscoveryNode(int clientEntityId, String configFile,
            String tdnid, String host, String port, String protocol,
            String output) {

      configFilePath = configFile;

      ServicePropertiesFactory servicePropertiesFactory = ServicePropertiesFactory
               .getInstance();
      serviceProperties = servicePropertiesFactory
               .getServiceProperties(clientEntityId);

      // MUST INITIALIZE THE CertificateManager
      CertificateManager certMan = new CertificateManager();

      String keystore = null;
      try {
         keystore = SystemInit.getInitData(SystemInit.SYS_KEYSTORE);
      } catch (SystemNotInitializedException e1) {
         log.error("", e1);
      }

      certMan.init(keystore, null);
      rootPubKey = CertificateUtil.getPublicKey(certMan, certMan.ROOT_CA_ALIAS);

      if (benchmark) {
         try {
            fout = new FileOutputStream(output);
         } catch (FileNotFoundException e) {
            log.error("", e);
         }
      }

      TDN_ID = tdnid;
      TDN_Credentials = CertificateUtil.getCertificate(certMan, TDN_ID);
      privateKey = CertificateUtil.getPrivateKey(certMan, TDN_ID);

      brokerHost = host;
      brokerPort = port;
      brokerConnProtocol = protocol;

      storedTopicAds = new Hashtable();

      TDN_TOPIC = Topics.TDN_TOPIC_BASE + "/" + TDN_ID;
      init();

      // Create the clock
      String nbHome = null;
      try {
         nbHome = SystemInit.getInitData(SystemInit.SYS_NB_HOME);
      } catch (SystemNotInitializedException e) {
         log.error("", e);
      }

      hrClock = HRClock.getClock(nbHome);
   }

   /**
    * Subscribes to various streams
    */
   private void init() {
      Properties prop = new Properties();

      // IMP: Might conflict w/ other ppl's entity IDs Maintain constant for
      // now
      int entityId = (new Random()).nextInt();
      prop.put("hostname", brokerHost);
      prop.put("portnum", brokerPort);

      System.out.println("[TopicDiscoveryNode] - Connecting using "
               + brokerConnProtocol + " @ " + brokerHost + ":" + brokerPort
               + " EntityID[" + entityId + "]");

      try {
         ClientService clientService = SessionService.getClientService(
                  entityId, configFilePath);
         clientService.initializeBrokerCommunications(prop, brokerConnProtocol);
         consumer = clientService.createEventConsumer(this);

         // Subscribe to my TDN_DISCOVERY
         Profile profile_TDN_DISCOVERY = clientService.createProfile(
                  TemplateProfileAndSynopsisTypes.STRING, Topics.TDN_DISCOVERY);
         consumer.subscribeTo(profile_TDN_DISCOVERY);

         // Subscribe to my TDN_TOPIC
         Profile profile_TDN_TOPIC = clientService.createProfile(
                  TemplateProfileAndSynopsisTypes.STRING, TDN_TOPIC);
         consumer.subscribeTo(profile_TDN_TOPIC);

         // Subscribe to TOPIC_ADS for storing advertisements
         Profile profile_TOPIC_ADS = clientService.createProfile(
                  TemplateProfileAndSynopsisTypes.STRING, Topics.TOPIC_ADS);
         consumer.subscribeTo(profile_TOPIC_ADS);

         // Subscribe to TOPIC_DISCO for listening and servicing Topic
         // discovery requests
         Profile profile_TOPIC_DISCO = clientService.createProfile(
                  TemplateProfileAndSynopsisTypes.STRING,
                  Topics.TOPIC_DISCOVERY_REQUEST);
         consumer.subscribeTo(profile_TOPIC_DISCO);

         // Create a producer
         producer = clientService.createEventProducer();
         producer.setSuppressRedistributionToSource(true);
      } catch (ServiceException e) {
         log.error("", e);
      }
   }

   /**
    * Send the specified message on specified topic
    * 
    * @param topicName -
    *           The topic on which to send the message
    * @param request -
    *           The message to be sent
    */
   public void send(String topicName, byte[] request) {
      producer.generateEventIdentifier(true);
      producer.setTemplateId(12345);
      producer.setDisableTimestamp(false);
      try {
         NBEvent nbEvent = producer.generateEvent(
                  TemplateProfileAndSynopsisTypes.STRING, topicName, request);
         producer.publishEvent(nbEvent);
         log.info("sent " + request.length + " bytes ON <" + topicName + ">");
      } catch (ServiceException e) {
         log.error("", e);
      }
   }

   private void handleTDNDiscoveryRequest(TDNDiscoveryRequest tdnDiscoReq) {
      log.info("TDNDiscoveryRequest from [" + tdnDiscoReq.getEntityId() + "]");

      long start = 0, stop = 0;
      // BENCHMARK
      if (benchmark) start = hrClock.getTimeMicroseconds();

      Certificate creds = tdnDiscoReq.getCredentials();
      if (checkEntityCredentials(creds)) {

         // BENCHMARK
         if (benchmark) {
            stop = hrClock.getTimeMicroseconds();
            try {
               fout.write(("TDNDisco#CheckCreds:" + (stop - start) + "\n")
                        .getBytes());
            } catch (IOException e) {
               log.error("", e);
            }
         }

         log.info("<" + tdnDiscoReq.getEntityId()
                  + "> Credentials verified !! Processing Request...");

         // For replying back
         String entityId = tdnDiscoReq.getEntityId();

         // TODO: This imfo is perhaps constant over the lifetime of TDN.
         // Make this constant rather than creating everytime

         TDNDiscoveryResponse tdnDiscoResp = new TDNDiscoveryResponse();
         tdnDiscoResp.setTDN_ID(TDN_ID);
         tdnDiscoResp.setCredentials(TDN_Credentials);

         byte[] tdnDiscoRespBytes = tdnDiscoResp.getBytes();

         byte[] response = new byte[1 + tdnDiscoRespBytes.length];
         response[0] = EventPayloadHeader.EVENT_TDN_DISCOVERY_RESPONSE;
         System.arraycopy(tdnDiscoRespBytes, 0, response, 1,
                  tdnDiscoRespBytes.length);

         // Send an encrypted message using Entity's public key.. so only
         // Entity can decipher the message

         start = hrClock.getTimeMicroseconds();

         EncryptedMessagePayload emp = EncryptedMessagePayload.generate(
                  response, creds.getPublicKey());

         // BENCHMARK
         if (benchmark) {
            stop = hrClock.getTimeMicroseconds();
            try {
               fout.write(("TDNDisco#EncryptResp:" + (stop - start) + "\n")
                        .getBytes());
            } catch (IOException e) {
               log.error("", e);
            }
         }
         // DISCARD ON ERROR, Entity'll timeout
         if (emp == null) return;

         send(Topics.ENTITY_TOPIC_BASE + "/" + entityId, emp.getBytes());
      } // ELSE Don't bother replying... just discard the request...!
   }

   private void handleTopicCreationRequest(TopicCreationRequest topicCreationReq) {
      log.info("Topic Creation request from: ["
               + topicCreationReq.getEntityId() + "]");

      long start = 0, stop = 0;

      if (benchmark) start = hrClock.getTimeMicroseconds();

      // Verify Entity Credentials ??
      if (!checkEntityCredentials(topicCreationReq.getCertificate())) return;

      // BENCHMARK
      if (benchmark) {
         stop = hrClock.getTimeMicroseconds();
         try {
            fout.write(("TopicCreation#CheckCreds:" + (stop - start) + "\n")
                     .getBytes());
         } catch (IOException e) {
            log.error("", e);
         }
      }
      // OK, Entity is good.. so process the request now...

      log.info("<" + topicCreationReq.getEntityId()
               + "> Credentials verified !! Processing Request...");

      start = hrClock.getTimeMicroseconds();

      String topicUUID = UUIDRetriever.getInstance().getTimeBasedUUIDAsString();

      TopicAdvertisement topicAd = new TopicAdvertisement(topicCreationReq);
      topicAd.setUuid(topicUUID);

      // Synopsis = UUID::Synopsis
      topicAd.setTopicSynopsis(topicUUID + "::" + topicAd.getTopicSynopsis());

      // TEST-CASE: Wrong synopsis (UNComment following Line for testing)
      // topicAd.setTopicSynopsis(topicUUID + ":: _GARBAGE_SYNOPSIS_");

      byte[] topicAdBytes = topicAd.getBytes();

      byte[] response = new byte[1 + topicAdBytes.length];
      response[0] = EventPayloadHeader.EVENT_TOPIC_CREATION_RESPONSE;
      System.arraycopy(topicAdBytes, 0, response, 1, topicAdBytes.length);

      log.info("Topic: " + topicAd.getTopicSynopsis());

      // BENCHMARK
      if (benchmark) {
         stop = hrClock.getTimeMicroseconds();
         try {
            fout.write(("TopicCreation#GenResp:" + (stop - start) + "\n")
                     .getBytes());
         } catch (IOException e) {
            log.error("", e);
         }
      }
      // Send an encrypted message using Entity's public key.. so only Entity
      // can decipher the message
      if (benchmark) start = hrClock.getTimeMicroseconds();

      EncryptedMessagePayload emp = EncryptedMessagePayload.generate(response,
               topicCreationReq.getCertificate().getPublicKey());

      // BENCHMARK
      if (benchmark) {
         stop = hrClock.getTimeMicroseconds();
         try {
            fout.write(("TopicCreation#EncryptResp:" + (stop - start) + "\n")
                     .getBytes());
         } catch (IOException e) {
            log.error("", e);
         }
      }
      // DISCARD ON ERROR, Entity'll timeout
      if (emp == null) return;

      if (benchmark) start = hrClock.getTimeMicroseconds();

      send(Topics.ENTITY_TOPIC_BASE + "/" + topicCreationReq.getEntityId(), emp
               .getBytes());

      // BENCHMARK
      if (benchmark) {
         stop = hrClock.getTimeMicroseconds();
         try {
            fout.write(("TopicCreation#SendResp:" + (stop - start) + "\n")
                     .getBytes());
         } catch (IOException e) {
            log.error("", e);
         }
      }
   }

   private void handleTopicAd(SignedTopicAdvertisement signedTopicAd) {
      if (!hasExpired(signedTopicAd.getTopicAd().getLifetime())) {
         String key = signedTopicAd.getTopicAd().getUuid();
         log.info("Storing Topic Ad: " + key);
         storedTopicAds.put(key, signedTopicAd);
      } else
         log.warn("Dropping Expired TopicAd !");
   }

   private void handleTopicDiscoveryRequest(TopicDiscoveryRequest req) {
      log.info("Topic Discovery Request !");

      long start = 0, stop = 0;

      if (benchmark) start = hrClock.getTimeMicroseconds();

      Certificate creds = req.getClientCredentials();

      if (!checkEntityCredentials(creds)) {
         log.warn("Discarding Topic discovery request. Invalid credentials !");
         return;
      }

      // BENCHMARK
      if (benchmark) {
         stop = hrClock.getTimeMicroseconds();
         try {
            fout.write(("TopicDisco#CheckCreds:" + (stop - start) + "\n")
                     .getBytes());
         } catch (IOException e) {
            log.error("", e);
         }
      }

      log.info("Credentials verified !! Processing Request...");

      TopicDiscoveryResponse resp = new TopicDiscoveryResponse();

      // Do the query and find the matching topic
      // TODO: add support for detailed querying...
      // Currently only quick match !
      if (benchmark) start = hrClock.getTimeMicroseconds();

      String requestor = ((X509Certificate) creds).getSubjectDN().getName();

      for (Enumeration e = storedTopicAds.elements(); e.hasMoreElements();) {
         SignedTopicAdvertisement sta = (SignedTopicAdvertisement) e
                  .nextElement();
         TopicAdvertisement topicAd = sta.getTopicAd();

         // Check if the entry is expired...
         if (hasExpired(topicAd.getLifetime())) {
            log.warn("Topic AD has Expired !");
            // Remove this TopicAd from the table
            storedTopicAds.remove(topicAd.getUuid());
            continue;
         }

         // Now match the query with the topic's synopsis
         if (evaluateQueryAgainstTopicSynopsis(topicAd, req)) {
            log.debug("MATCHED... !");

            // HG: added Aug 15, 2006
            // NOW, check if the requestor is allowed to discover this topic

            if (topicAd.getAuthorizedUsers().contains(requestor))
               resp.add(sta);
            else {
               log
                        .debug("[" + requestor
                                 + "] NOT allowed to discover topic...");
            }
         }
      }

      // BENCHMARK
      if (benchmark) {
         stop = hrClock.getTimeMicroseconds();
         try {
            fout.write(("TopicDisco#Discovery:" + (stop - start) + "\n")
                     .getBytes());
         } catch (IOException e) {
            log.error("", e);
         }
      }
      log.info(resp.getMatchingTopicCount() + " Matching Topics found !");

      // Send response only if AT-LEAST 1 mathcing entry was found !!
      if (resp.getMatchingTopicCount() > 0) {

         // Send an encrypted message using Entity's public key.. so only
         // Entity
         // can decipher the message
         if (benchmark) start = hrClock.getTimeMicroseconds();

         EncryptedMessagePayload emp = EncryptedMessagePayload.generate(resp
                  .getBytes(), creds.getPublicKey());

         // BENCHMARK
         if (benchmark) {
            stop = hrClock.getTimeMicroseconds();
            try {
               fout.write(("TopicDisco#EncryptResp:" + (stop - start) + "\n")
                        .getBytes());
            } catch (IOException e) {
               log.error("", e);
            }
         }
         // DISCARD ON ERROR, Entity'll timeout
         if (emp == null) return;

         send(Topics.TOPIC_DISCOVERY_RESPONSE, emp.getBytes());
      }
   }

   /**
    * Checks the current time to see if the specified time has expired
    * 
    * @param cal -
    *           The time to check
    * @return TRUE if current time > specified time, FALSE otherwise
    */
   private boolean hasExpired(Calendar cal) {
      Calendar currentTime = Calendar.getInstance();
      return currentTime.getTimeInMillis() > cal.getTimeInMillis();
   }

   /**
    * Verifies the supplied Entity Credentials
    * 
    * @param creds -
    *           Supplied Credentials
    * @return TRUE if the entity is verified, FALSE otherwise
    */
   private boolean checkEntityCredentials(Certificate creds) {
      try {
         creds.verify(rootPubKey);
         return true;
      } catch (InvalidKeyException e) {
         log.error("", e);
      } catch (CertificateException e) {
         log.error("", e);
      } catch (NoSuchAlgorithmException e) {
         log.error("", e);
      } catch (NoSuchProviderException e) {
         log.error("", e);
      } catch (SignatureException e) {
         log.error("", e);
      }
      return false;
   }

   /**
    * Evaluates the specified query against the stored Topic Ads.
    * 
    * @param topicAd -
    *           topic Ad against which the query is to be evaluated
    * @param req -
    *           The specified request
    * @return TRUE if the Ad matches the request, FALSE otherwise
    */
   private boolean evaluateQueryAgainstTopicSynopsis(
            TopicAdvertisement topicAd, TopicDiscoveryRequest req) {

      // Get the topic synopsis. Strip the leading 'UUID::' (36 + 2 chars)
      String topicSynopsis = topicAd.getTopicSynopsis().substring(38);

      log.debug("Matching [" + req.getQuery() + "] with <" + topicSynopsis
               + ">");

      switch (req.getSearchType()) {
         case Topics.MATCHING_STRING:
            log.debug("String Matching...");
            return req.getQuery().equals(topicSynopsis);

         case Topics.MATCHING_INTEGER:
            log.debug("Int Matching...");
            int topicnum = Integer.parseInt(topicSynopsis);
            int reqestedNum = Integer.parseInt(req.getQuery());
            return topicnum == reqestedNum;

         case Topics.MATCHING_REGEX:
            log.debug("REGEX Matching...");
            Pattern pattern = Pattern.compile(req.getQuery());
            Matcher matcher = pattern.matcher(topicSynopsis);
            matcher.reset();
            return matcher.find();

         // NOT YET DEFINED !
         case Topics.MATCHING_TAGVALUE:
         break;
         case Topics.MATCHING_XPATH:
         break;
      }

      return false;
   }

   /**
    * Process events
    */
   public synchronized void onEvent(NBEvent nbEvent) {
      long start = 0, stop = 0;

      String topic = (String) nbEvent.getContentSynopsis();

      // This is a TDN Discovery request
      if (topic.equals(Topics.TDN_DISCOVERY)) {

         TDNDiscoveryRequest tdnDiscoReq = TDNDiscoveryRequest
                  .createObjectFromBytes(nbEvent.getContentPayload());

         if (tdnDiscoReq == null) {
            log.warn("{TDN_DISCOVERY}Garbage EVENT recieved !");
            return;
         }

         handleTDNDiscoveryRequest(tdnDiscoReq);
      }

      // This would be a TOPIC Creation Request
      else if (topic.equals(TDN_TOPIC)) {

         EncryptedMessagePayload emp = EncryptedMessagePayload
                  .createObjectFromBytes(nbEvent.getContentPayload());

         byte[] topicCreationReqBytes = emp.getDecryptedPayload(privateKey);

         if (topicCreationReqBytes == null) {
            log.error("Unable to decrypt TopicCreationRequest: ");
            return; // If error occurs.. discard request
         }

         TopicCreationRequest topicCreationReq = TopicCreationRequest
                  .createObjectFromBytes(topicCreationReqBytes);

         if (topicCreationReq == null) {
            log.warn("{TDN_TOPIC}Garbage EVENT recieved !");
            return;
         }

         handleTopicCreationRequest(topicCreationReq);
      }

      // Store the Ad in the hashtable
      else if (topic.equals(Topics.TOPIC_ADS)) {

         if (benchmark) start = hrClock.getTimeMicroseconds();
         boolean digestVerified = false;

         SignedTopicAdvertisement signedTopicAd = SignedTopicAdvertisement
                  .createObjectFromBytes(nbEvent.getContentPayload());

         // Step 1: First validate sender's public key
         Certificate cert = signedTopicAd.getOwnerCertificate();

         if (checkEntityCredentials(cert))
            digestVerified = signedTopicAd.verify();

         if (!digestVerified) {
            log.warn("PHONY Message recieved. ");
            return;
         } else {
            log.info("Signature Verified !");
         }

         handleTopicAd(signedTopicAd);

         // BENCHMARK
         if (benchmark) {
            stop = hrClock.getTimeMicroseconds();
            try {
               fout.write(("StoreTOPICAd:" + (stop - start) + "\n").getBytes());
            } catch (IOException e1) {
               log.error("", e1);
            }
         }

      }

      // Topic discovery request
      else if (topic.equals(Topics.TOPIC_DISCOVERY_REQUEST)) {
         // BENCHMARK
         if (benchmark) start = hrClock.getTimeMicroseconds();

         TopicDiscoveryRequest req = TopicDiscoveryRequest
                  .createObjectFromBytes(nbEvent.getContentPayload());

         if (req == null) {
            log.warn("{TOPIC_DISCOVERY_REQUEST}Garbage EVENT recieved !");
            return;
         }
         handleTopicDiscoveryRequest(req);

         // BENCHMARK
         if (benchmark) {
            stop = hrClock.getTimeMicroseconds();
            try {
               fout.write(("TOPICDISCO:" + (stop - start) + "\n").getBytes());
            } catch (IOException e1) {
               log.error("", e1);
            }
         }
      }
   }

   /**
    * Starts the Topic Discovery node. Use -help for usage<br>
    * TODO: Add parsing of command line parameters to provide more fine grained
    * control
    */
   public static void main(String[] args) {

      if (args[0].equals("-help")) {
         System.out.println("USAGE: ");
         System.out
                  .println("ARG_0: <hostnameToUse> OR -auto (for using broker discovery");
         System.out.println("ARG_1: <port>");
         System.out.println("ARG_2: <protocol>");
         System.out.println("-----------------------");
         System.exit(0);
      }

      SystemInit.init();

      String host = "localhost";
      String port = "25000";
      String protocol = "niotcp";

      String serviceConfigPath = null;
      try {
         serviceConfigPath = SystemInit
                  .getInitData(SystemInit.SYS_SERVICE_CONFIG_PATH);
      } catch (SystemNotInitializedException e) {
         e.printStackTrace();
      }

      if (args.length > 0) {
         host = args[0];
         if (host.equals("-auto")) {
            // invoke Broker Discovery
            // USE Broker Discovery to locate broker !
            // BROKER DISCOVERY:---

            BrokerDiscoveryHelper bdh = new BrokerDiscoveryHelper(
                     (new Random()).nextInt(), serviceConfigPath, 5000, 1, 1);

            BrokerDiscoveryResponse[] ba = bdh.discover(
                     new String[] { protocol }, "**", "*");

            Properties props = (Properties) ba[0].brokerInfo
                     .getSupportedTransportProtocols().get("niotcp");

            host = ba[0].brokerInfo.getHostname();
            port = props.getProperty("NIOTCPServerPort");
         }
         // ---
         else {
            if (args.length > 1) port = args[1];
            if (args.length > 2) protocol = args[2];
         }
      }

      System.out.println("Connecting to " + host + ":" + port + " [" + protocol
               + "]");

      // ---

      // TopicDiscoveryNode TDN = new TopicDiscoveryNode(
      // 2000,
      // "/home/hgadgil/projects/NB1rc1/config/ServiceConfiguration.txt",
      // "tdn", host, port, protocol, "/tmp/tdn.out");

      TopicDiscoveryNode TDN = new TopicDiscoveryNode(2000, serviceConfigPath,
               "tdn", host, port, protocol, "c:/tdn.out");

   }
}